home *** CD-ROM | disk | FTP | other *** search
/ Cream of the Crop 26 / Cream of the Crop 26.iso / os2 / pvm34b3.zip / pvm34b3 / pvm3 / pvmgs / pvmgsu_aux.c < prev    next >
C/C++ Source or Header  |  1997-07-22  |  33KB  |  1,079 lines
  1.  
  2. static char rcsid[] =
  3.     "$Id: pvmgsu_aux.c,v 1.6 1997/07/09 13:51:36 pvmsrc Exp $";
  4.  
  5. /*
  6.  *         PVM version 3.4:  Parallel Virtual Machine System
  7.  *               University of Tennessee, Knoxville TN.
  8.  *           Oak Ridge National Laboratory, Oak Ridge TN.
  9.  *                   Emory University, Atlanta GA.
  10.  *      Authors:  J. J. Dongarra, G. E. Fagg, M. Fischer
  11.  *          G. A. Geist, J. A. Kohl, R. J. Manchek, P. Mucci,
  12.  *         P. M. Papadopoulos, S. L. Scott, and V. S. Sunderam
  13.  *                   (C) 1997 All Rights Reserved
  14.  *
  15.  *                              NOTICE
  16.  *
  17.  * Permission to use, copy, modify, and distribute this software and
  18.  * its documentation for any purpose and without fee is hereby granted
  19.  * provided that the above copyright notice appear in all copies and
  20.  * that both the copyright notice and this permission notice appear in
  21.  * supporting documentation.
  22.  *
  23.  * Neither the Institutions (Emory University, Oak Ridge National
  24.  * Laboratory, and University of Tennessee) nor the Authors make any
  25.  * representations about the suitability of this software for any
  26.  * purpose.  This software is provided ``as is'' without express or
  27.  * implied warranty.
  28.  *
  29.  * PVM version 3 was funded in part by the U.S. Department of Energy,
  30.  * the National Science Foundation and the State of Tennessee.
  31.  */
  32.  
  33. /*
  34.  *    pvmgsu_aux.c - auxiliary group library routines, gather and scatter  
  35.  *       6 Jun 1995     Native mode reduce for Paragon. Donato
  36.  *       8 Mar 1994     Added reduce & assoc routines. Donato & P.Papadopoulos
  37.  *      24 Apr 1994     Added scatter, gather, gs_get_datasize routines. Donato
  38.  *
  39.  */
  40.  
  41. #ifdef HASSTDLIB
  42. #include <stdlib.h>
  43. #endif
  44. #include <stdio.h>
  45. #include <pvm3.h>
  46. #include "pvmalloc.h"
  47. #include "bfunc.h"
  48. #include "lpvm.h"
  49. #include <pvmtev.h>
  50. #include "tevmac.h"
  51. #include "global.h"
  52. #include "pvmmimd.h"
  53. #include "pvmgsd.h"
  54.  
  55. extern int pvm_errno;
  56. extern int pvmmytid;
  57. extern int pvmtoplvl;
  58. extern struct Pvmtracer pvmtrc;
  59.  
  60.  
  61.  
  62. /* ==========                                         ================ 
  63.  * ========== Declarations & Routines for the Paragon ================
  64.  * ==========                                         ================ 
  65.  * Note:  int and long are the same on the paragon 
  66.  */
  67.  
  68. #if defined(IMA_PGON)
  69.  
  70. extern int pvmpgonpartsize;           /* from lpvmmimd.c */
  71.  
  72. /* ================ NativeFunction() for the PGON ====================     */
  73. /* This routine compares the address of the user provided function
  74.    to the built-in pvm functions to determine and then call the
  75.    appropriate native mode function on the Paragon.
  76. */
  77. int
  78. NativeFunction(user_func, datatype, x, y, num)
  79. #ifdef IMA_SCO
  80.   void (*user_func)(int*, void*, void*, int*, int*);
  81. #else
  82.   void (*user_func)();
  83. #endif
  84. void *x, *y;
  85. int  datatype, num;
  86. {
  87.     if (user_func==PvmSum)
  88.     {
  89.         switch(datatype)
  90.         {
  91.             case (PVM_INT):
  92.             case (PVM_LONG):
  93.                 _gisum((int *) x, num, (int *) y);
  94.                 break;
  95.             case (PVM_FLOAT):
  96.                 _gssum((float *) x, num, (float *) y);
  97.                 break;
  98.             case (PVM_DOUBLE):
  99.                 _gdsum((double *) x, num, (double *) y);
  100.                 break;
  101.             default:
  102.                 return(PvmNotImpl);
  103.         }   /* end switch */
  104.     }
  105.     else
  106.     if (user_func==PvmMax)
  107.     {
  108.         switch(datatype)
  109.         {
  110.             case (PVM_INT):
  111.             case (PVM_LONG):
  112.                 _gihigh((int *) x, num, (int *) y);
  113.                 break;
  114.             case (PVM_FLOAT):
  115.                 _gshigh((float *) x, num, (float *) y);
  116.                 break;
  117.             case (PVM_DOUBLE):
  118.                 _gdhigh((double *) x, num, (double *) y);
  119.                 break;
  120.             default:
  121.                 return(PvmNotImpl);
  122.         }   /* end switch */
  123.     }
  124.     else
  125.     if (user_func==PvmMin)
  126.     {
  127.         switch(datatype)
  128.         {
  129.             case (PVM_INT):
  130.             case (PVM_LONG):
  131.                 _gilow((int *) x, num, (int *) y);
  132.                 break;
  133.             case (PVM_FLOAT):
  134.                 _gslow((float *) x, num, (float *) y);
  135.                 break;
  136.             case (PVM_DOUBLE):
  137.                 _gdlow((double *) x, num, (double *) y);
  138.                 break;
  139.             default:
  140.                 return(PvmNotImpl);
  141.         }   /* end switch */
  142.     }
  143.     else
  144.     if (user_func==PvmProduct)
  145.     {
  146.         switch(datatype)
  147.         {
  148.             case (PVM_INT):
  149.             case (PVM_LONG):
  150.                 _giprod((int *) x, num, (int *) y);
  151.                 break;
  152.             case (PVM_FLOAT):
  153.                 _gsprod((float *) x, num, (float *) y);
  154.                 break;
  155.             case (PVM_DOUBLE):
  156.                 _gdprod((double *) x, num, (double *) y);
  157.                 break;
  158.             default:
  159.                 return(PvmNotImpl);
  160.         }  /* end switch */
  161.     }
  162.     else
  163.         return(PvmNotImpl);
  164.  
  165.     return(PvmOk);
  166.  
  167. }  /* end NativeFunction() */
  168.  
  169.  
  170. #endif /* #ifdef PGON */
  171.  
  172.  
  173.  
  174. /* ================ pvm_reduce() =====================================     */
  175. /*
  176.   int info = pvm_reduce(void (*func)(), void *data, int count, int datatype,
  177.                         int msgtag, char *gname, int rootinst)
  178.   where
  179.       void (*func)(int *datatype, void *data, void *work, int *num, int *info)
  180.  
  181.   Currently, this implementation uses a fan in algorithm to perform
  182.   the reduce operation.
  183.  
  184.   Each group member sends their data to the coordinator on their host machine.
  185.   The coordinator performs the specified function combining its own data and 
  186.   the data from the group members on the same host.  
  187.   
  188.   On the Paragon the nx native global operations are utilized if all the
  189.   nodes of the Paragon are part of the group.
  190.  
  191.   The coordinators then pass their results on to the specified root node 
  192.   of the reduce operation.
  193. */
  194.  
  195. int pvm_reduce(func, data, count, datatype, msgtag, gname, rootinst)
  196. #ifdef  IMA_SCO
  197.     void (*func)(int*, void*, void*, int*, int*);
  198. #else
  199.     void (*func)();
  200. #endif
  201.     void *data;
  202.     int count, datatype, msgtag, rootinst;
  203.     char *gname; 
  204. {
  205.     int cnt, roottid, datasize, cc=PvmOk, rbuf, sbuf;
  206.     int coordinator, nmembers_on_host, nhosts_in_group, mask=0;
  207.     void *work = NULL;      /* work array to be allocated */
  208.  
  209.     int  (*packfunc)(), (*unpackfunc)();
  210.     int x;
  211.  
  212.     TEV_DECLS
  213.  
  214.     BGN_TRACE( TEV_REDUCE, gname, TEV_DID_MC, &msgtag );
  215.  
  216.     rbuf = pvm_setrbuf(0);            /* set receive buf */
  217.     sbuf = pvm_mkbuf(PvmDataDefault);
  218.     sbuf = pvm_setsbuf(sbuf);
  219.  
  220.     if ( (data == NULL) || (count <= 0) ) /* check some parameters */
  221.     {
  222.       cc = PvmBadParam;
  223.       goto done;
  224.     }
  225.   
  226.     /* get instance number - caller must be in group, root must be in group */
  227.     if ( (cc = pvm_getinst(gname, pvmmytid))            < PvmOk ) goto done;
  228.     if ( (cc = roottid  = pvm_gettid (gname, rootinst)) < PvmOk ) goto done;
  229.     if ( (cc = datasize = gs_get_datasize(datatype))    < PvmOk ) goto done;
  230.  
  231.     /* set up pointers to the appropriate pack and unpack routines */
  232.     if ( (cc = gs_pack_unpack(datatype, &packfunc, &unpackfunc) ) < PvmOk)
  233.         goto done;
  234.  
  235.     if ((work = (void *) PVM_ALLOC(count*datasize, "pvm_reduce")) == NULL)
  236.     {
  237.         cc = PvmNoMem;
  238.         goto done;
  239.     }
  240.  
  241.     pvm_grphostinfo(gname, gs_tidtohost(pvmmytid), &coordinator, 
  242.                     &nmembers_on_host, &nhosts_in_group);
  243.  
  244. #if defined(IMA_PGON)
  245.     /*  if all the nodes are participating,  
  246.         then call the Native mode version, if one exists.
  247.         The native function call is a side-effect of the call to NativeFunction.
  248.         If the coordinator and roottid are PGON nodes, there is no
  249.         need to differentiate (e.g. pass data from coordinator to roottid).
  250.     */
  251.  
  252.     mask = TIDHOST | TIDPTYPE;
  253.  
  254.     if ( TIDISNODE(pvmmytid) && 
  255.        ( pvmpgonpartsize == nmembers_on_host ) &&
  256.        ( (cc = NativeFunction(func, datatype, data, work, count)) == PvmOk) )
  257.     {
  258.         if ( TIDISNODE(roottid) && (pvmmytid & mask) == (roottid & mask) )
  259.             coordinator = roottid;
  260.     }
  261.     else
  262. #endif    /* end if defined(IMA_PGON) */
  263.     {
  264.         if ( (pvmmytid==coordinator) && (nmembers_on_host>1) )
  265.         {
  266.             /* recv data from other group members on same host, perform func */
  267.             for (cnt = nmembers_on_host-1; cnt>0; cnt--)
  268.             {
  269.                 if ( (cc = pvm_recv(-1, msgtag) )          < PvmOk) goto done;
  270.                 if ( (cc = (*unpackfunc)( work, count, 1)) < PvmOk) goto done;
  271.                 (*func)( &datatype, data, work, &count, &cc );
  272.                 if (cc < PvmOk) goto done;           /* error flag from func */
  273.             }  
  274.         }
  275.         else if (pvmmytid != coordinator)
  276.         {
  277.             /* send data to the data coordinator on this same host */
  278.             pvm_initsend(PvmDataDefault);
  279.             if ( (cc = (*packfunc)( data, count, 1) )  < PvmOk ) goto done;
  280.             if ( (cc = pvm_send( coordinator, msgtag)) < PvmOk ) goto done;;
  281.         }
  282.     } 
  283.  
  284.     if ( (pvmmytid==coordinator) && (pvmmytid != roottid) )
  285.     {
  286.         /* send data to the roottid for the reduce operation */
  287.         pvm_initsend(PvmDataDefault);
  288.         if ( (cc = (*packfunc)( data, count, 1) )  < PvmOk ) goto done;
  289.         if ( (cc = pvm_send( roottid, msgtag)) < PvmOk ) goto done;
  290.     } 
  291.  
  292.     /* Root node of the reduce operation: 
  293.        - get data from my host coordinator, if necessary
  294.        - get work values from each of the other hosts
  295.        - perform the specified functions on data from other hosts
  296.     */
  297.     if (pvmmytid == roottid) 
  298.     {
  299.         /* if root isn't the host coordinator, receive from coordinator 1st */
  300.         if  (pvmmytid != coordinator) 
  301.         {
  302.             if ( (cc = pvm_recv(coordinator, msgtag) ) < PvmOk) goto done;
  303.             if ( (cc = (*unpackfunc)( data, count, 1)) < PvmOk) goto done;
  304.         }
  305.  
  306.         if (nhosts_in_group-- <=  0) goto done;
  307.  
  308.         /* recv data from other group members on diff host, perform func */
  309.         for (cnt = nhosts_in_group; cnt>0; cnt--)
  310.         {
  311.             if ( (cc = pvm_recv(-1, msgtag) )          < PvmOk) goto done;
  312.             if ( (cc = (*unpackfunc)( work, count, 1)) < PvmOk) goto done;
  313.             (*func)( &datatype, data, work, &count, &cc );
  314.             if (cc < PvmOk) goto done;               /* error flag from func */
  315.         }   /* end for */
  316.     }
  317.     
  318.     cc = PvmOk;
  319.     
  320.   done:
  321.   
  322.     /* restore user's buffers */
  323.     pvm_freebuf(pvm_setrbuf(rbuf));
  324.     pvm_freebuf(pvm_setsbuf(sbuf));
  325.  
  326.     if (work != NULL) PVM_FREE(work);         /* free work space  */
  327.  
  328.     if (cc < 0) lpvmerr("pvm_reduce",cc);
  329.   
  330.     END_TRACE( TEV_REDUCE, TEV_DID_CC, &cc );
  331.  
  332.     return(cc);
  333.  
  334. } /* end pvm_reduce() */
  335.  
  336.  
  337.  
  338. /* ================ gs_pack_unpack()==================================     */
  339. /*  
  340.     int info = gs_pack_unpack( int datatype, 
  341.                                int (**packfunc)(), int (**unpackfunc)() )
  342.  
  343.     Sets up pointers to the appropriate pack and unpack function based
  344.     on datatype specified.
  345. */
  346.  
  347. int 
  348. gs_pack_unpack(datatype, packfunc, unpackfunc)
  349. int datatype, (**packfunc)(), (**unpackfunc)();
  350. {
  351.  
  352.     switch(datatype) 
  353.     {
  354.         case (PVM_STR):  
  355.             *packfunc = pvm_pkstr; 
  356.             *unpackfunc = pvm_upkstr; 
  357.             break;
  358.         case (PVM_BYTE):  
  359.             *packfunc = pvm_pkbyte; 
  360.             *unpackfunc = pvm_upkbyte; 
  361.             break;
  362.         case (PVM_SHORT):
  363.             *packfunc = pvm_pkshort; 
  364.             *unpackfunc = pvm_upkshort; 
  365.             break;
  366.         case (PVM_INT):
  367.             *packfunc = pvm_pkint;
  368.             *unpackfunc = pvm_upkint; 
  369.             break;
  370.         case (PVM_LONG):      
  371.             *packfunc = pvm_pklong; 
  372.             *unpackfunc = pvm_upklong; 
  373.             break;
  374.         case (PVM_FLOAT):
  375.             *packfunc = pvm_pkfloat; 
  376.             *unpackfunc = pvm_upkfloat; 
  377.             break;
  378.         case (PVM_DOUBLE):
  379.             *packfunc = pvm_pkdouble; 
  380.             *unpackfunc = pvm_upkdouble; 
  381.             break;
  382.         case (PVM_CPLX):
  383.             *packfunc = pvm_pkcplx; 
  384.             *unpackfunc = pvm_upkcplx; 
  385.             break;
  386.         case (PVM_DCPLX):
  387.             *packfunc = pvm_pkdcplx; 
  388.             *unpackfunc = pvm_upkdcplx; 
  389.             break;
  390.         default:
  391.             return(PvmBadParam);
  392.     }
  393.  
  394.     return(PvmOk);
  395. }   
  396. /* ================ PvmMax()==========================================     */
  397. /* 
  398.   void PvmMax(int *datatype, void *x, void *y, int *num, int *info)
  399.   Assigns the elements of x the maximum value between the
  400.   corresponding elements of x and y.  
  401.   For complex values the maximum is determined by maximum modulus.
  402. */
  403.  
  404. void 
  405. PvmMax(datatype, x, y, num, info)
  406. int *datatype;
  407. void *x, *y; 
  408. int *num, *info;
  409. {
  410.     char   *xchar,   *ychar;
  411.     int    *xint,    *yint;
  412.     short  *xshort,  *yshort;
  413.     long   *xlong,   *ylong;
  414.     float  *xfloat,  *yfloat;
  415.     double *xdouble, *ydouble;
  416.     float   xfreal, xfimag, yfreal, yfimag;
  417.     double  xdreal, xdimag, ydreal, ydimag;
  418.     float   xsqrfloat, ysqrfloat;
  419.     double  xsqrdouble, ysqrdouble;
  420.     
  421.     int i, count;
  422.   
  423.     count = *num;
  424.   
  425.     switch(*datatype) 
  426.     {
  427.         case (PVM_BYTE):
  428.             xchar = (char *) x;
  429.             ychar = (char *) y;
  430.             for (i=0; i<count; i++) xchar[i] = MAX(xchar[i], ychar[i]);
  431.             break;
  432.         case (PVM_SHORT):
  433.             xshort = (short *) x;
  434.             yshort = (short *) y;
  435.             for (i=0; i<count; i++) xshort[i] = MAX(xshort[i], yshort[i]);
  436.             break;
  437.         case (PVM_INT):
  438.             xint = (int *) x;
  439.             yint = (int *) y;
  440.             for (i=0; i<count; i++) xint[i] = MAX(xint[i], yint[i]);
  441.             break;
  442.         case (PVM_LONG):
  443.             xlong = (long *) x;
  444.             ylong = (long *) y;
  445.             for (i=0; i<count; i++) xlong[i] = MAX(xlong[i], ylong[i]);
  446.             break;
  447.         case (PVM_FLOAT):
  448.             xfloat = (float *) x;
  449.             yfloat = (float *) y;
  450.             for (i=0; i<count; i++) xfloat[i] = MAX(xfloat[i], yfloat[i]);
  451.             break;
  452.         case (PVM_DOUBLE):
  453.             xdouble = (double *) x;
  454.             ydouble = (double *) y;
  455.             for (i=0; i<count; i++) xdouble[i] = MAX(xdouble[i], ydouble[i]);
  456.             break;
  457.         case (PVM_CPLX):
  458.             /* complex - complex*8 in fortran - treated as two floats */
  459.             /* returns the complex pair with the greatest magnitude */
  460.             xfloat = (float *) x;
  461.             yfloat = (float *) y;
  462.             for (i=0; i<2*count; i+=2)
  463.             {
  464.                 xfreal = xfloat[i];
  465.                 xfimag = xfloat[i+1];
  466.                 yfreal = yfloat[i];
  467.                 yfimag = yfloat[i+1];
  468.                 xsqrfloat = xfreal*xfreal + xfimag*xfimag;
  469.                 ysqrfloat = yfreal*yfreal + yfimag*yfimag;
  470.                 if (ysqrfloat > xsqrfloat)
  471.                 {
  472.                     xfloat[i]   = yfreal;
  473.                     xfloat[i+1] = yfimag;
  474.                 }
  475.             }
  476.             break;
  477.         case (PVM_DCPLX):
  478.             /* double complex - complex*16 in fortran - treated as 2 doubles */
  479.             /* returns the complex pair with the greatest magnitude */
  480.             xdouble = (double *) x;
  481.             ydouble = (double *) y;
  482.             for (i=0; i<2*count; i+=2)
  483.             {
  484.                 xdreal = xdouble[i];
  485.                 xdimag = xdouble[i+1];
  486.                 ydreal = ydouble[i];
  487.                 ydimag = ydouble[i+1];
  488.                 xsqrdouble = xdreal*xdreal + xdimag*xdimag;
  489.                 ysqrdouble = ydreal*ydreal + ydimag*ydimag;
  490.                 if (ysqrdouble > xsqrdouble)
  491.                 {
  492.                     xdouble[i]   = ydreal;
  493.                     xdouble[i+1] = ydimag;
  494.                 }
  495.             }
  496.             break;
  497.         default:
  498.             *info = PvmBadParam;
  499.             return;
  500.     }  /* end switch */
  501.  
  502.     *info = PvmOk;
  503.     return;
  504.  
  505. }  /* end of PvmMax() */
  506.  
  507.  
  508.  
  509. /* ================ PvmMin()==========================================     */
  510. /* 
  511.   void PvmMin(int *datatype, void *x, void *y, int *num, int *info)
  512.  
  513.   Assigns the elements of x the minimum value between the
  514.   corresponding elements of x and y.
  515.   For complex values the minimum is determined by minimum modulus.
  516.  
  517. */
  518.  
  519. void 
  520. PvmMin(datatype, x, y, num, info)
  521. int *datatype;
  522. void *x, *y;
  523. int  *num, *info;
  524. {
  525.     char   *xchar,   *ychar;
  526.     short  *xshort,  *yshort;
  527.     int    *xint,    *yint;
  528.     long   *xlong,   *ylong;
  529.     float  *xfloat,  *yfloat;
  530.     double *xdouble, *ydouble;
  531.     float   xfreal, xfimag, yfreal, yfimag;
  532.     double  xdreal, xdimag, ydreal, ydimag;
  533.     float   xsqrfloat, ysqrfloat;
  534.     double  xsqrdouble, ysqrdouble;
  535.   
  536.     int i, count;
  537.   
  538.     count = *num;
  539.   
  540.     switch(*datatype) 
  541.     {
  542.         case (PVM_BYTE):
  543.             xchar = (char *) x;
  544.             ychar = (char *) y;
  545.             for (i=0; i<count; i++) xchar[i] = MIN(xchar[i], ychar[i]);
  546.             break;
  547.         case (PVM_SHORT):
  548.             xshort = (short *) x;
  549.             yshort = (short *) y;
  550.             for (i=0; i<count; i++) xshort[i] = MIN(xshort[i], yshort[i]);
  551.             break;
  552.         case (PVM_INT):
  553.             xint = (int *) x;
  554.             yint = (int *) y;
  555.             for (i=0; i<count; i++) xint[i] = MIN(xint[i], yint[i]);
  556.             break;
  557.         case (PVM_LONG):
  558.             xlong = (long *) x;
  559.             ylong = (long *) y;
  560.             for (i=0; i<count; i++) xlong[i] = MIN(xlong[i], ylong[i]);
  561.             break;
  562.         case (PVM_FLOAT):
  563.             xfloat = (float *) x;
  564.             yfloat = (float *) y;
  565.             for (i=0; i<count; i++) xfloat[i] = MIN(xfloat[i], yfloat[i]);
  566.             break;
  567.         case (PVM_DOUBLE):
  568.             xdouble = (double *) x;
  569.             ydouble = (double *) y;
  570.             for (i=0; i<count; i++) xdouble[i] = MIN(xdouble[i], ydouble[i]);
  571.             break;
  572.         case (PVM_CPLX):
  573.             /* complex - complex*8 in fortran - treated as two floats */
  574.             /* returns the complex pair with the smaller magnitude */
  575.             xfloat = (float *) x;
  576.             yfloat = (float *) y;
  577.             for (i=0; i<2*count; i+=2)
  578.             {
  579.                 xfreal = xfloat[i];
  580.                 xfimag = xfloat[i+1];
  581.                 yfreal = yfloat[i];
  582.                 yfimag = yfloat[i+1];
  583.                 xsqrfloat = xfreal*xfreal + xfimag*xfimag;
  584.                 ysqrfloat = yfreal*yfreal + yfimag*yfimag;
  585.                 if (ysqrfloat < xsqrfloat)
  586.                 {
  587.                     xfloat[i]   = yfreal;
  588.                     xfloat[i+1] = yfimag;
  589.                 }
  590.             }
  591.             break;
  592.         case (PVM_DCPLX):
  593.             /* double complex - complex*16 in fortran - treated as 2 doubles */
  594.             /* returns the complex pair with the smaller magnitude */
  595.             xdouble = (double *) x;
  596.             ydouble = (double *) y;
  597.             for (i=0; i<2*count; i+=2)
  598.             {
  599.                 xdreal = xdouble[i];
  600.                 xdimag = xdouble[i+1];
  601.                 ydreal = ydouble[i];
  602.                 ydimag = ydouble[i+1];
  603.                 xsqrdouble = xdreal*xdreal + xdimag*xdimag;
  604.                 ysqrdouble = ydreal*ydreal + ydimag*ydimag;
  605.                 if (ysqrdouble < xsqrdouble)
  606.                 {
  607.                     xdouble[i]   = ydreal;
  608.                     xdouble[i+1] = ydimag;
  609.                 }
  610.             }
  611.             break;
  612.         default:
  613.             *info = PvmBadParam;
  614.             return;
  615.     }  /* end switch */
  616.      
  617.     *info = PvmOk;
  618.     return;
  619.  
  620. }  /* end of PvmMin() */
  621.  
  622.  
  623. /* ================ PvmSum()==========================================     */
  624.  
  625. /* 
  626.   void PvmSum(int *datatype, void *x, void *y, *num, *info)
  627.  
  628.   Assigns the elements of x the sum of the corresponding elements of x and y.
  629. */
  630.  
  631. void 
  632. PvmSum(datatype, x, y, num, info)
  633. int *datatype;
  634. void *x, *y;
  635. int *num, *info;
  636. {
  637.     short  *xshort,  *yshort;
  638.     int    *xint,    *yint;
  639.     long   *xlong,   *ylong;
  640.     float  *xfloat,  *yfloat;
  641.     double *xdouble, *ydouble;
  642.   
  643.     int i, count;
  644.   
  645.     count = *num;
  646.   
  647.     switch(*datatype) 
  648.     {
  649.         case (PVM_SHORT):
  650.             xshort = (short *) x;
  651.             yshort = (short *) y;
  652.             for (i=0; i<count; i++) xshort[i] += yshort[i];
  653.             break;
  654.         case (PVM_INT):
  655.             xint = (int *) x;
  656.             yint = (int *) y;
  657.             for (i=0; i<count; i++) xint[i] += yint[i];
  658.             break;
  659.         case (PVM_LONG):
  660.             xlong = (long *) x;
  661.             ylong = (long *) y;
  662.             for (i=0; i<count; i++) xlong[i] += ylong[i];
  663.             break;
  664.         case (PVM_FLOAT):
  665.             xfloat = (float *) x;
  666.             yfloat = (float *) y;
  667.             for (i=0; i<count; i++) xfloat[i] += yfloat[i];
  668.             break;
  669.         case (PVM_DOUBLE):
  670.             xdouble = (double *) x;
  671.             ydouble = (double *) y;
  672.             for (i=0; i<count; i++) xdouble[i] += ydouble[i];
  673.             break;
  674.         case (PVM_CPLX):
  675.             /* complex - complex*8 in fortran - treated as two floats */
  676.             /* returns the sum of the two complex pairs */
  677.             xfloat = (float *) x;
  678.             yfloat = (float *) y;
  679.             for (i=0; i<2*count; i++) xfloat[i]  += yfloat[i];
  680.             break;
  681.         case (PVM_DCPLX):
  682.             /* double complex - complex*16 in fortran - treated as 2 doubles */
  683.             /* returns the sum of the two complex pairs */
  684.             xdouble = (double *) x;
  685.             ydouble = (double *) y;
  686.             for (i=0; i<2*count; i++) xdouble[i]   += ydouble[i];
  687.             break;
  688.         default:
  689.             *info = PvmBadParam;
  690.             return;
  691.     }  /* end switch */
  692.  
  693.     *info = PvmOk;
  694.     return;
  695.  
  696. }  /* end of PvmSum() */
  697.  
  698.  
  699. /* ================ PvmProduct()======================================     */
  700. /* 
  701.   void PvmProduct(int *datatype, void *x, void *y, *num, *info)
  702.  
  703.   Assigns the elements of x the sum of the corresponding elements of x and y.
  704. */
  705.  
  706. void 
  707. PvmProduct(datatype, x, y, num, info)
  708. int *datatype;
  709. void *x, *y;
  710. int *num, *info;
  711. {
  712.     short  *xshort,  *yshort;
  713.     int    *xint,    *yint;
  714.     long   *xlong,   *ylong;
  715.     float  *xfloat,  *yfloat, a,b,c,d;
  716.     double *xdouble, *ydouble, da,db,dc,dd;
  717.   
  718.     int i, count;
  719.   
  720.     count = *num;
  721.   
  722.     switch(*datatype) 
  723.     {
  724.         case (PVM_SHORT):
  725.             xshort = (short *) x;
  726.             yshort = (short *) y;
  727.             for (i=0; i<count; i++) xshort[i] *= yshort[i];
  728.             break;
  729.         case (PVM_INT):
  730.             xint = (int *) x;
  731.             yint = (int *) y;
  732.             for (i=0; i<count; i++) xint[i] *= yint[i];
  733.             break;
  734.         case (PVM_LONG):
  735.             xlong = (long *) x;
  736.             ylong = (long *) y;
  737.             for (i=0; i<count; i++) xlong[i] *= ylong[i];
  738.             break;
  739.         case (PVM_FLOAT):
  740.             xfloat = (float *) x;
  741.             yfloat = (float *) y;
  742.             for (i=0; i<count; i++) xfloat[i] *= yfloat[i];
  743.             break;
  744.         case (PVM_DOUBLE):
  745.             xdouble = (double *) x;
  746.             ydouble = (double *) y;
  747.             for (i=0; i<count; i++) xdouble[i] *= ydouble[i];
  748.             break;
  749.         case (PVM_CPLX):
  750.             /* complex - complex*8 in fortran - treated as two floats */
  751.             /* returns the product of the two complex pairs */
  752.             xfloat = (float *) x;
  753.             yfloat = (float *) y;
  754.             for (i=0; i<2*count; i += 2) 
  755.             {
  756.                 a = xfloat[i]; b = xfloat[i+1], c = yfloat[i]; d = yfloat[i+1];
  757.                 xfloat[i] = a*c - b*d;
  758.                 xfloat[i+1] = a*d + b*c;
  759.             }
  760.             break;
  761.         case (PVM_DCPLX):
  762.             /* double complex - complex*16 in fortran - treated as 2 doubles */
  763.             /* returns the  product of the two complex pairs */
  764.             xdouble = (double *) x;
  765.             ydouble = (double *) y;
  766.             for (i=0; i<2*count; i+= 2) 
  767.             {
  768.                 da = xdouble[i]; db = xdouble[i+1], 
  769.                 dc = ydouble[i]; dd = ydouble[i+1];
  770.                 xdouble[i] = da*dc - db*dd;
  771.                 xdouble[i+1] = da*dd + db*dc;
  772.             }
  773.             break;
  774.         default:
  775.             *info = PvmBadParam;
  776.             return;
  777.     }  /* end switch */
  778.  
  779.     *info = PvmOk;
  780.     return;
  781.  
  782. }  /* end of PvmProduct() */
  783.  
  784.  
  785. /* ================ pvm_gather()======================================     */
  786. /*    
  787.   int info = pvm_gather(void *result, void *data, int count, int datatype, 
  788.                         int msgtag,  char *gname, int rootinst)
  789.  
  790.   Performs a gather of messages from each member of the group
  791.   to a specified member of the group.
  792.  
  793.   Each member of the group 'gname' sends a message 'data' 
  794.   of type 'datatype' and length 'count' to the root member of the group.
  795.   The root receives these messages into a single array 'result'
  796.   which is of length, at least, (number of group members)*'count'.
  797.   The values received from the ith member of the group are
  798.   placed into the 'result' array starting at position i*'count'.
  799.   The root member of the group is specified by its instance number,
  800.   'rootginst', in that group.
  801. */
  802.  
  803. int 
  804. pvm_gather(result, data, count, datatype, msgtag, gname, rootinst)
  805. void *result, *data;
  806. int  count, datatype, msgtag, rootinst;
  807. char *gname;
  808. {
  809.     int mytid, roottid, myginst, datasize, gsize, *tids = 0, i, cc;
  810.     int sbuf, rbuf;
  811.   
  812.     int (*packfunc)(), (*unpackfunc)();  /* ptrs to pack and unpack functions */
  813.     int x;
  814.   
  815.     TEV_DECLS
  816.  
  817.     BGN_TRACE( TEV_GATHER, gname, TEV_DID_MC, &msgtag );
  818.  
  819.     if ( (data == NULL) || (count <= 0) ) /* check some parameters */
  820.     {
  821.         cc = PvmBadParam;
  822.         goto done;
  823.     }
  824.   
  825.     /* set up pointers to the appropriate pack and unpack routines */
  826.     if ( (cc = gs_pack_unpack(datatype, &packfunc, &unpackfunc) ) < 0)
  827.         goto done;
  828.   
  829.     /* root must be member of the group */
  830.     if ( (cc = roottid = pvm_gettid(gname,rootinst)) < 0 ) goto done;
  831.   
  832.     /* get instance number - caller must be in group */
  833.     if ( (cc = myginst = pvm_getinst(gname, pvmmytid)) < 0 ) goto done;
  834.  
  835.     if (myginst == rootinst)     /* I am the root for the gather operation */
  836.     {
  837.         if ( result == NULL) /* check result parameter */
  838.         {
  839.             cc = PvmBadParam;
  840.             goto done;
  841.         }
  842.   
  843.         /* get the number of bytes per element of type datatype */
  844.         if ( (cc = datasize = gs_get_datasize(datatype)) < 0  ) goto done;
  845.   
  846.         /* Get the list of tids.  These must be contiguous (no holes). */
  847.         if ( (cc = gs_get_tidlist(gname, msgtag, &gsize, &tids, 1)) < 0)
  848.             goto done;
  849.   
  850.         rbuf = pvm_setrbuf(0);
  851.   
  852.         /* Get the values, put them in the correct place in the result. 
  853.            The instance numbers should be contiguous within the group.
  854.         */
  855.         for (i=0; i<gsize; i++)
  856.         {
  857.             /* The root copies its data into its result array */
  858.             if (i == myginst) 
  859.             {
  860.                 BCOPY((char *) data, (char *) result + i*datasize*count, 
  861.                       datasize*count);
  862.             }
  863.             else
  864.             {
  865.                 if ( (cc = pvm_recv( tids[i], msgtag )) < 0 )
  866.                 {
  867.                     pvm_freebuf(pvm_setrbuf(rbuf)); /* restore user recv buf*/
  868.                     goto done;
  869.                 }
  870.                 if ((cc=(*unpackfunc)((char *)result+i*datasize*count,count,1))<0)
  871.                 {
  872.                     pvm_freebuf(pvm_setrbuf(rbuf)); /* restore user recv buf*/
  873.                     goto done;
  874.                 }
  875.   
  876.             } /* end if (i == myginst) */
  877.   
  878.         } /* end for-loop */
  879.   
  880.         pvm_freebuf(pvm_setrbuf(rbuf)); /* restore user's receive buf */
  881.     }
  882.     else  
  883.     {    /* everyone except the root sends data to the root */
  884.         sbuf = pvm_mkbuf(PvmDataDefault);
  885.         sbuf = pvm_setsbuf(sbuf);
  886.      
  887.         if ( (cc = (*packfunc)( data, count, 1)) < 0)
  888.         {
  889.           pvm_freebuf(pvm_setsbuf(sbuf)); /* restore user's send buf */
  890.           goto done;
  891.         }
  892.         if ( (cc = pvm_send( roottid, msgtag)) < 0)
  893.         {
  894.           pvm_freebuf(pvm_setsbuf(sbuf)); /* restore user's send buf */
  895.           goto done;
  896.         }
  897.         pvm_freebuf(pvm_setsbuf(sbuf)); /* restore user's send buf */
  898.   
  899.     }  /* end if-else */
  900.  
  901.  
  902.   cc = PvmOk;
  903.  
  904. done:
  905.     if (tids) free(tids);
  906.     
  907.     if (cc < 0) lpvmerr("pvm_gather",cc);
  908.   
  909.     END_TRACE( TEV_GATHER, TEV_DID_CC, &cc );
  910.  
  911.     return(cc);
  912.  
  913. }  /* end pvm_gather() */
  914.  
  915. /* ================ pvm_scatter()=====================================     */
  916. /*    
  917.   int info = pvm_scatter(void *result, int *data,  int count, int datatype, 
  918.                          void msgtag, char *gname, int rootinst)
  919.  
  920.   Performs a scatter of messages from the specified root member of the
  921.   group to each of the members of the group.
  922.  
  923.   Each member of the group 'gname' receives a message 'result' 
  924.   of type 'datatype' and length 'count' from the root member of the group.
  925.   The root sends these messages from a single array 'data'
  926.   which is of length, at least, (number of group members)*'count'.
  927.   The values sent to the ith member of the group are
  928.   taken from the 'data' array starting at position i*'count'.
  929.   The root member of the group is specified by its instance number,
  930.   'rootginst', in that group.
  931. */
  932.  
  933. int 
  934. pvm_scatter(result, data, count, datatype, msgtag, gname, rootinst)
  935. void *result, *data;
  936. int  count, datatype, msgtag, rootinst;
  937. char *gname;
  938. {
  939.     int roottid, myginst, datasize, gsize, *tids = 0, i, cc;
  940.     int sbuf, rbuf;
  941.  
  942.     int (*packfunc)(), (*unpackfunc)();  /* ptrs to pack and unpack functions */
  943.     int x;
  944.  
  945.     TEV_DECLS
  946.  
  947.     BGN_TRACE( TEV_SCATTER, gname, TEV_DID_MC, &msgtag );
  948.  
  949.     if ( (result == NULL) || (count <= 0) ) /* check some parameters */
  950.     {
  951.         cc = PvmBadParam;
  952.         goto done;
  953.     }
  954.  
  955.     /* set up pointers to the appropriate pack and unpack routines */
  956.     if ( (cc = gs_pack_unpack(datatype, &packfunc, &unpackfunc)) < 0)
  957.         goto done;
  958.  
  959.     /* root must be member of the group */
  960.     if ( (roottid = pvm_gettid(gname,rootinst)) < 0 )
  961.     {
  962.         cc = roottid;
  963.         goto done;
  964.     }
  965.  
  966.     /* get instance number - caller must be in group */
  967.     if ( (cc = myginst = pvm_getinst(gname, pvmmytid)) < 0 ) goto done;
  968.  
  969.     /* I am the root node for the scatter operation */
  970.     if (myginst == rootinst)
  971.     {
  972.         if ( data == NULL) /* check data parameter */
  973.         {
  974.             cc = PvmBadParam;
  975.             goto done;
  976.         }
  977.  
  978.         /* get the number of bytes per element of type datatype */
  979.         if ( (cc = datasize = gs_get_datasize(datatype)) < 0  ) goto done;
  980.   
  981.         /* Get the list of tids.  These must be contiguous (no holes). */
  982.         if ( (cc = gs_get_tidlist(gname, msgtag, &gsize, &tids, 1)) < 0)
  983.             goto done;
  984.   
  985.         sbuf = pvm_mkbuf(PvmDataDefault);
  986.   
  987.         /* The root sends values to everyone, except itself, in the group.
  988.            For itself, the root copies the data into its result array.
  989.         */
  990.         for (i=0; i<gsize; i++)
  991.         {
  992.             if (i == myginst)
  993.               BCOPY((char *) data + i*datasize*count, (char *) result,
  994.                     datasize*count);
  995.             else
  996.             {
  997.                 sbuf = pvm_initsend(PvmDataDefault);
  998.                 if ((cc=(*packfunc)((char *)data+i*datasize*count,count,1))<0)
  999.                 {
  1000.                 pvm_freebuf(pvm_setsbuf(sbuf));   /* restore user's sendbuf */
  1001.                 goto done;
  1002.                 }
  1003.                 if ( (cc = pvm_send( tids[i], msgtag)) < 0)
  1004.                 {
  1005.                     pvm_freebuf(pvm_setsbuf(sbuf)); /* restore user's sendbuf */
  1006.                     goto done;
  1007.                 }
  1008.             } /* end if-else */
  1009.         } /* end for-loop */
  1010.         pvm_freebuf(pvm_setsbuf(sbuf)); /* restore user's send buf */
  1011.     }
  1012.     else
  1013.     {
  1014.         /* everyone receives a result from the root, except the root */
  1015.         rbuf = pvm_setrbuf(0);
  1016.         if ( (cc = pvm_recv( roottid, msgtag )) < 0)
  1017.         {
  1018.             pvm_freebuf(pvm_setrbuf(rbuf)); /* restore user's receive buf */
  1019.             goto done;
  1020.         }
  1021.         if ( (cc = (*unpackfunc)( result, count, 1)) < 0)
  1022.         {
  1023.             pvm_freebuf(pvm_setrbuf(rbuf)); /* restore user's receive buf */
  1024.             goto done;
  1025.         }
  1026.         pvm_freebuf(pvm_setrbuf(rbuf)); /* restore user's receive buf */
  1027.     
  1028.     }  /* end if-else */
  1029.     
  1030.     cc = PvmOk;
  1031.  
  1032. done:
  1033.     if (tids) free(tids);
  1034.   
  1035.     if (cc < 0) lpvmerr("pvm_scatter",cc);
  1036.  
  1037.     END_TRACE( TEV_SCATTER, TEV_DID_CC, &cc);
  1038.  
  1039.     return(cc);
  1040.  
  1041. }  /* end pvm_scatter() */
  1042.  
  1043. /* ================ gs_get_datasize() ================================     */
  1044. /*    
  1045.   int info = gs_get_datasize(int datatype)
  1046.  
  1047.   Returns the size in bytes of a single element of type datatype.
  1048. */
  1049.  
  1050. int 
  1051. gs_get_datasize(datatype)
  1052. int datatype;
  1053. {
  1054.  
  1055.     switch (datatype)
  1056.     {
  1057.         case (PVM_STR):
  1058.         case (PVM_BYTE):
  1059.             return(sizeof(char));
  1060.         case (PVM_SHORT):
  1061.             return(sizeof(short));
  1062.         case (PVM_INT):
  1063.             return(sizeof(int));
  1064.         case (PVM_LONG):
  1065.             return(sizeof(long));
  1066.         case (PVM_FLOAT):
  1067.             return(sizeof(float));
  1068.         case (PVM_DOUBLE):
  1069.             return(sizeof(double));
  1070.         case (PVM_CPLX):
  1071.             return(2*sizeof(float));
  1072.         case (PVM_DCPLX):
  1073.             return(2*sizeof(double));
  1074.         default:
  1075.             return(PvmBadParam);
  1076.     }  /* end switch (datatype) */
  1077.  
  1078. }  /* end gs_get_datasize() */
  1079.